package com.etl.component.oracle;

import cn.hutool.core.util.StrUtil;
import com.component.api.constants.ComponentCodeConstants;
import com.component.api.constants.ComponentConstants;
import com.component.api.model.PortData;
import com.component.api.model.ProcessResult;
import com.component.api.model.data.ColumnData;
import com.component.api.model.dto.DatabaseParamDTO;
import com.component.api.model.param.ParamPair;
import com.component.api.model.param.TableColumn;
import com.component.api.utils.SqlUtils;
import com.google.common.collect.Lists;
import com.common.utils.constant.SQLConstants;
import com.etl.component.common.AbstractFlinkComponent;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.types.Row;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 描述：
 * oracle数据库读取
 *  因为是单表 都采用jdbc读取
 * @author xianggj
 * @Date 2021/10/12 18:10
 **/
public class OracleReadComponent extends AbstractFlinkComponent {

    @Override
    public ProcessResult process(List<PortData> datas, List<ParamPair> paramPairs) {
        //让oracle强制遵循jdbc规范
        System.getProperties().setProperty("oracle.jdbc.J2EE13Compliant", "true");
        Map<String, Object> params = paramToObject(paramPairs);
        DatabaseParamDTO databaseParamDTO = (DatabaseParamDTO)params.get(ComponentConstants.CONFIG);
        String driver = databaseParamDTO.getDriver();
        String url = databaseParamDTO.getUrl();
        String userName = databaseParamDTO.getUserName();
        String pwd = databaseParamDTO.getPwd();
        String tableName = databaseParamDTO.getTableName();
        List<TableColumn> columns = (List<TableColumn>)params.get(ComponentConstants.COLUMNS);
        columns = columns.stream().filter(e->StrUtil.isNotBlank(e.getColumnName())).collect(Collectors.toList());
        TypeInformation[] fieldTypes =new TypeInformation[columns.size()];
        String[] fieldNames = new String[columns.size()];
        StringBuffer sql = new StringBuffer(SQLConstants.SELECT);
        List<String> colSqlStr = new ArrayList<>();
        int i = 0;
        for (TableColumn tc : columns) {
            String colName = tc.getColumnName();
            fieldNames[i] = colName;
            colName = SqlUtils.addEscape(colName , SQLConstants.ORACLE_ESC);
            String alias = tc.getAlias();
            if (StrUtil.isNotBlank(alias)){
                colName = colName + SQLConstants.AS + SqlUtils.addEscape(alias , SQLConstants.ORACLE_ESC);
            }
            colSqlStr.add(colName);
            String type = tc.getDataType();
            fieldTypes[i] = typeToBasicTypeInfo(tc);
            i++;
        }
        sql.append(StrUtil.join(",", colSqlStr));
        sql.append(SQLConstants.FROM);
        sql.append(SqlUtils.addEscape(tableName , SQLConstants.ORACLE_ESC));

        RowTypeInfo rowTypeInfo =new RowTypeInfo(fieldTypes, fieldNames);
        JDBCInputFormat jdbcInputFormat = JDBCInputFormat.buildJDBCInputFormat()
                .setDrivername(driver)
                .setDBUrl(url)
                .setUsername(userName)
                .setPassword(pwd)
                .setQuery(sql.toString())
                .setRowTypeInfo(rowTypeInfo)
                .finish();

        final ExecutionEnvironment env = getEnv();
        DataSource s = env.createInput(jdbcInputFormat);
        BatchTableEnvironment tableEnv = BatchTableEnvironment.create(env);
        //视图名称
        String viewName = "temp_" + tableName;
        tableEnv.createTemporaryView( viewName,s);
//        tableEnv.sqlQuery("select * from test").printSchema();
        Table query = tableEnv.sqlQuery(SQLConstants.SELECT_All + viewName);
        DataSet result = tableEnv.toDataSet(query, Row.class);
        return execute(result, columns, "执行oracle读取操作");
    }


    @Override
    public List<PortData> previewOutputColumns(List<PortData> datas, List<ParamPair> paramPairs) {
        Map<String, Object> params = paramToObject(paramPairs);
        List<TableColumn> columns = (List<TableColumn>)params.get(ComponentConstants.COLUMNS);
        return Lists.newArrayList(new PortData(getRule(), new ColumnData(columns)));
    }

    @Override
    public String getRule() {
        return ComponentCodeConstants.ORACLE_INPUT;
    }


    /**
     * 字符串类型转BasicTypeInfo
     * @param tc
     */
    public TypeInformation typeToBasicTypeInfo(TableColumn tc ) {
        String type = tc.getDataType();
        type = StrUtil.isBlank(type) ? "varchar2" : type.toLowerCase();
        switch (type) {
            case "decimal":
            case "number":
                return BasicTypeInfo.BIG_DEC_TYPE_INFO;
            case "int":
                return BasicTypeInfo.INT_TYPE_INFO;
            case "float":
                return BasicTypeInfo.FLOAT_TYPE_INFO;
            case "double":
                return BasicTypeInfo.DOUBLE_TYPE_INFO;
            case "date":
            case "timestamp":
            case "timestamp(6)":
                return SqlTimeTypeInfo.TIMESTAMP;
            default:
                return BasicTypeInfo.STRING_TYPE_INFO;
        }

    }

}
